package com.zy.hadoop;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FsUrlStreamHandlerFactory;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.junit.jupiter.api.Test;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.URL;

/**
 * @program: hadoop-practice
 * @className: HdfsApiTest.java
 * @description:
 * @author: ZY
 * @create: 2020年10月13日 11:47
 **/

public class HdfsApiTest {

    /**
     * 通过URL的方式访问hadoop文件
     * @throws IOException 异常
     */
    @Test
    public void urlAccess() throws IOException {
        // 1.注册url
        URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
        // 2.获取hdfs文件的输入流
        InputStream inputStream = new URL("hdfs://192.168.10.101:8020/dir1/start-dfs.sh").openStream();
        // 3.获取本地文件的输出流
        FileOutputStream outputStream = new FileOutputStream(new File("D://a.sh"));
        // 4.实现文件的拷贝
        IOUtils.copy(inputStream, outputStream);
        // 5.关流
        IOUtils.closeQuietly(inputStream);
        IOUtils.closeQuietly(outputStream);
        System.out.println("从hadoop中下载文件成功");
    }

    /**
     * 方式1：通过设置默认文件系统来获取分布式文件系统
     * @throws IOException 异常
     */
    @Test
    public void getFileSystem1() throws IOException {
        Configuration configuration = new Configuration();
        // 指定使用的文件系统类型(不设置的话，默认使用本地文件系统)
        configuration.set("fs.defaultFS", "hdfs://192.168.10.101:8020");
        // 获取指定的文件系统
        FileSystem fileSystem = FileSystem.get(configuration);
        System.out.println(fileSystem);
    }

    /**
     * 方式2：通过URI直接获取分布式文件系统
     * @throws Exception 异常
     */
    @Test
    public void getFileSystem2() throws Exception {
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.10.101:8020"), new Configuration());
        System.out.println(fileSystem);
    }

    /**
     * 方式3：通过newInstance获取分布式文件系统
     * @throws Exception 异常
     */
    @Test
    public void getFileSystem3() throws Exception {
        Configuration configuration = new Configuration();
        configuration.set("fs.defaultFS", "hdfs://192.168.10.101:8020");
        FileSystem fileSystem = FileSystem.newInstance(configuration);
        System.out.println(fileSystem);
    }

    /**
     * 方式4：通过newInstance获取分布式文件系统
     * @throws Exception 异常
     */
    @Test
    public void getFileSystem4() throws Exception {
        Configuration configuration = new Configuration();
        FileSystem fileSystem = FileSystem.newInstance(new URI("hdfs://192.168.10.101:8020"), configuration);
        System.out.println(fileSystem);
    }

    /**
     * 遍历HDFS中所有文件
     */
    @Test
    public void listFiles() throws Exception {
        // 获取分布式文件系统
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.10.101:8020"), new Configuration());
        // 获取RemoteIterator得到所有的文件夹或者文件
        RemoteIterator<LocatedFileStatus> fileIterator = fileSystem.listFiles(new Path("/"), true);
        while (fileIterator.hasNext()) {
            LocatedFileStatus locatedFile = fileIterator.next();
            System.out.println("文件全路径：" + locatedFile.getPath() + " 文件名：" + locatedFile.getPath().getName());
            System.out.println("文件block数：" + locatedFile.getBlockLocations().length);
            System.out.println("文件副本数：" + locatedFile.getReplication());
            System.out.println();
        }
        fileSystem.close();
    }

    /**
     * 在HDFS中创建文件夹
     * @throws Exception 异常
     */
    @Test
    public void mkdirs() throws Exception {
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.10.101:8020"), new Configuration());
        boolean mkdirs = fileSystem.mkdirs(new Path("/hello/hadoop/mydir"));
        System.out.println("文件夹是否创建成功：" + mkdirs);
        fileSystem.close();
    }

    /**
     * 在HDFS中创建文件
     * @throws Exception 异常
     */
    @Test
    public void creatFile() throws Exception {
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.10.101:8020"), new Configuration());
        fileSystem.create(new Path("/aa/bb/cc/文本.txt"));
        fileSystem.close();
        System.out.println("文件创建成功");
    }

    /**
     * 方式1：从HDFS下载文件
     * @throws Exception 异常
     */
    @Test
    public void downloadFile1() throws Exception {
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.10.101:8020"), new Configuration());
        FSDataInputStream inputStream = fileSystem.open(new Path("/dir1/start-dfs.sh"));
        FileOutputStream outputStream = new FileOutputStream(new File("D:\\aqt.sh"));
        IOUtils.copy(inputStream, outputStream);
        IOUtils.closeQuietly(inputStream);
        IOUtils.closeQuietly(outputStream);
        fileSystem.close();
        System.out.println("下载成功");
    }

    /**
     * 方式2：从HDFS下载文件
     * @throws Exception 异常
     */
    @Test
    public void downloadFile2() throws Exception {
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.10.101:8020"), new Configuration());
        fileSystem.copyToLocalFile(new Path("/dir1/start-dfs.sh"), new Path("D:\\aqttt.sh"));
        fileSystem.close();
        System.out.println("下载成功");
    }

    /**
     * 向HDFS上传文件
     * @throws Exception 异常
     */
    @Test
    public void uploadFile() throws Exception {
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.10.101:8020"), new Configuration());
        fileSystem.copyFromLocalFile(new Path("file:///D:\\application-prod.yml"), new Path("/hello/hadoop/mydir"));
        fileSystem.close();
        System.out.println("上传文件成功");
    }

    /**
     * 将本地小文件合并成大文件上传到HDFS
     * @throws Exception 异常
     */
    @Test
    public void uploadFiles() throws Exception {
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.10.101:8020"), new Configuration());
        FSDataOutputStream outputStream = fileSystem.create(new Path("/aa/bb/cc/bigFile1.xml"));
        LocalFileSystem local = FileSystem.getLocal(new Configuration());
        FileStatus[] files = local.listStatus(new Path("D:\\files"));
        for (FileStatus file : files) {
            FSDataInputStream inputStream = local.open(file.getPath());
            IOUtils.copy(inputStream, outputStream);
            IOUtils.closeQuietly(inputStream);
        }
        IOUtils.closeQuietly(outputStream);
        fileSystem.close();
        System.out.println("合并上传成功");
    }

}
